package com.cps.platform.service.hbase.spi;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import javax.ws.rs.Path;
import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;

import com.cps.platform.core.BaseService;
import com.cps.platform.core.Constant;
import com.cps.platform.core.Mapbean;
import com.cps.platform.core.hbase.FilterBean;
import com.cps.platform.core.hbase.HbasePut;
import com.cps.platform.core.json.JsonParserUtil;
import com.cps.platform.core.util.ReadProperties;
import com.cps.platform.core.util.ValidateUtil;
import com.cps.platform.service.hbase.IHbaseService;

/**
 * <p>
 * 文件名称: DefaultHbaseService.java
 * </p>
 * <p>
 * 文件描述: HBASE的rest服务的默认实现
 * <p>
 * 版权所有: 版权所有(C)2011
 * </p>
 * <p>
 * 公 司: 联通系统集成有限公司
 * </p>
 * <p>
 * 完成日期：2012-12-3
 * </p>
 * 
 * @author ChenZhang
 */
@Component
@Path("/hbaseService")
public class DefaultHbaseService extends BaseService implements IHbaseService{

	private static Logger log = Logger.getLogger(DefaultHbaseService.class);
	private static Configuration cfg = null;
	private static HTablePool pool = null;
	
	//HBASE数据库的familyName定义，设定只使用一个familyName，因此
	private final String  DEFAULT_FAMILYNAME = "infoFamily";
	
	//HBASE数据库的列值，用于标识数据所属的租户，该值为保留属性
	private final String TENANTID = "paas_tenantId";
	
	DefaultHbaseService() {
		initConfig();
	}
	
	@Override
	protected void init(){
		try{
			ReadProperties readProperties = new ReadProperties("hbase.properties");
			cfg = HBaseConfiguration.create();
			// configuration.set("hbase.master", "192.168.0.201:60000");
			cfg.set(Constant.quornum,
					readProperties.readValue(Constant.quornum));
			cfg.set(Constant.clientPort,
					readProperties.readValue(Constant.clientPort));
			
			pool = new HTablePool(cfg, Integer.parseInt(readProperties.readValue(Constant.poolSize)));
			
			//初始化表
			int createdTableSign = Integer.parseInt(readProperties.readValue(Constant.created));
			if(createdTableSign!=1) {
				this.creatTable(Constant.paasconfig);
				
				String tablesJson = readProperties.readValue(Constant.hbaseTable);
				List<String> tables = JsonParserUtil.getObjectsFromJson(tablesJson, String.class);
				
				for(String table:tables) {
					this.creatTable(table);
				}
				
				readProperties.setProperty(Constant.created, "1");
				
				log.info("Init tables OK!");
			}
			log.info("HBaseConfiguration initial success.");
		}catch(Exception e) {
			log.error("HBaseConfiguration initial failed. The reason is "+ e.getMessage());
		}
	}
	
	@Override
	protected void setServiceId() {
		ReadProperties readProperties = new ReadProperties("hbase.properties");
		String servicesJson = readProperties.readValue(Constant.platformServices);
		Mapbean mapBean = (Mapbean) JsonParserUtil.getMapbean(servicesJson);
		super.serviceId = (String) mapBean.getMap().get(Constant.platformServicesId);
		super.authToken = (String) mapBean.getMap().get(Constant.platformServicesToken);
	}
	
	private Configuration getConfig(){
		return cfg;
	}
	
	private HTablePool getPool(){
		return pool;
	}
	
	@Override
	public Response creatTable(String appId, String authToken, String tenantId, String tablename) {
		if(ValidateUtil.isNull(appId, authToken, tenantId)){
			return Response.status(Status.BAD_REQUEST).build(); 
		}
		
		try {
			creatTable(tablename);
			
			return Response.ok(tablename).build();
		} catch (Exception e) {
			log.error("Failed to created "+ tablename +" tables. The reason is "+ e.getMessage());
			return Response.status(Status.INTERNAL_SERVER_ERROR).build();  
		}
	}
	
	/**
	 * 创建表
	 * @param tablename
	 * @throws Exception
	 */
	private void creatTable(String tablename) throws Exception {
		HBaseAdmin admin = null;
		admin = new HBaseAdmin(this.getConfig());
		if (admin.tableExists(tablename)) {
			log.info("Table "+ tablename +" Exists!!!");
		} else {
			HTableDescriptor tableDesc = new HTableDescriptor(tablename);
			tableDesc.addFamily(new HColumnDescriptor(DEFAULT_FAMILYNAME));
			admin.createTable(tableDesc);
			log.info("Successfully created "+ tablename +" tables");
		}
	}

	@Override
	public Response dropTable(String appId, String authToken, String tenantId, String tablename) {
		if(ValidateUtil.isNull(appId, authToken, tenantId)){
			return Response.status(Status.BAD_REQUEST).build(); 
		}
		
		try {
			HBaseAdmin admin = new HBaseAdmin(this.getConfig());
			if (admin.tableExists(tablename)) {
				admin.disableTable(tablename);
				admin.deleteTable(tablename);
			}
			log.info("Successfully droped "+ tablename +" tables.");
			return Response.ok(tablename).build();
		} catch (Exception e) {
			String message = "Failed to droped "+ tablename +" tables. The reason is "+ e.getMessage();
			log.error(message);
			return Response.status(Status.INTERNAL_SERVER_ERROR).build();  
		}
	}

	@Override
	public String addOrUpdateData(String appId, String authToken, String tenantId, String hbasePutJson) {
		if(ValidateUtil.isNull(appId, authToken, tenantId)){
			throw new WebApplicationException(Response.status(Status.BAD_REQUEST).build());
		}
		
		HbasePut hbasePut = null;
		if(ValidateUtil.isNull(hbasePutJson)){
			throw new WebApplicationException(Response.status(Status.BAD_REQUEST).build());
		}else{
			try{
				hbasePut = (HbasePut) JsonParserUtil.getObjectFromJson(hbasePutJson, HbasePut.class);
			}catch (Exception e) {
				log.error("Add or update data failure. The reason is "+ e.getMessage());
				throw new WebApplicationException(Response.status(Status.BAD_REQUEST).build());
			}
		}
		
		String tableName = hbasePut.getTableName();
		String row = hbasePut.getRowKey();
		
		try{
			Map<String, String> keyValues = hbasePut.getKeyValues();
			keyValues.put(this.TENANTID, tenantId);
			
			//HTable table = new HTable(cfg, tableName);
			HTableInterface table = this.getPool().getTable(tableName);
	
			Put put = new Put(new String(row).getBytes());
			// 写入信息
			Iterator<String> it = keyValues.keySet().iterator();
			while(it.hasNext()){
				String key = it.next();
				String value = keyValues.get(key);
				
				byte[] valueByte = null;
				if(value!=null && !"".equals(value.trim())) {
					valueByte = new String(value).getBytes();
				}
				
				put.add(new String(DEFAULT_FAMILYNAME).getBytes(),new String(key).getBytes(), valueByte);
			}
			
			table.put(put);
			table.flushCommits();
	
			putTable(table);
			log.info("Add or update data ["+tableName+"]["+row+"] success.");
			return JsonParserUtil.getJsonFromObject(hbasePut);
		} catch (Exception e) {
			String message = "Add or update data ["+tableName+"]["+row+"] failure. The reason is "+ e.getMessage();
			log.error(message);
			throw new WebApplicationException(Response.status(Status.INTERNAL_SERVER_ERROR).build());
		}
	}
	
	@Override
	public Response delDatas(String appId, String authToken, String tenantId, 
			String tablename, String rowKeysJson) {
		if(ValidateUtil.isNull(appId, authToken, tenantId)){
			return Response.status(Status.BAD_REQUEST).build(); 
		}
		
		try{
			HTableInterface table = this.getPool().getTable(tablename);
			// 删除
			List<Delete> deld = new ArrayList<Delete>();
			List<String> rowKeys = JsonParserUtil.getObjectsFromJson(rowKeysJson, String.class);
			for(String rowKey: rowKeys){
				deld.add(new Delete(new String(rowKey).getBytes()));
			}
			
			table.delete(deld);
			putTable(table);
			
			log.info("DEL data [tablename:"+tablename+",rowkeys:"+rowKeysJson+"] success.");
			return Response.ok(tablename+","+rowKeysJson).build();
		} catch (Exception e) {
			String message = "Delete data [tablename:"+tablename+",rowkeys:"+rowKeysJson+"] failure. The reason is "+ e.getMessage();
			log.error(message);
			return Response.status(Status.INTERNAL_SERVER_ERROR).build();  
		}
	}
	
	@Override
	public Response delData(String appId, String authToken, String tenantId, 
			String tablename, String rowKey) {
		if(ValidateUtil.isNull(appId, authToken, tenantId)){
			return Response.status(Status.BAD_REQUEST).build(); 
		}
		
		try{
			HTableInterface table = this.getPool().getTable(tablename);
			// 删除
			List<Delete> deld = new ArrayList<Delete>();
			deld.add(new Delete(new String(rowKey).getBytes()));
			
			table.delete(deld);
			putTable(table);
			
			log.info("DEL data [tablename:"+tablename+",rowkeys:"+rowKey+"] success.");
			return Response.ok(tablename+","+rowKey).build();
		} catch (Exception e) {
			String message = "Delete data [tablename:"+tablename+",rowkeys:"+rowKey+"] failure. The reason is "+ e.getMessage();
			log.error(message);
			return Response.status(Status.INTERNAL_SERVER_ERROR).build();  
		}
	}

	@Override
	public String selectRowKey(String appId, String authToken, String tenantId, 
			String tableName, String rowKey) {
		try {
			List<HbasePut> puts = selectRowKeysFamilyColumn(tableName, null, null, rowKey);
			if(puts.size()==1){
				return JsonParserUtil.getJsonFromObject(puts.get(0));
			}else{
				return null;
			}
		} catch (IOException e) {
			log.error("Query data [tablename:"+tableName+"rowKey:"+rowKey+"] failed. The reason is "+ e.getMessage());
			throw new WebApplicationException(Response.status(Status.INTERNAL_SERVER_ERROR).build());
		}
	}
	
	@Override
	public String selectAllData(String appId, String authToken, String tenantId, 
			String tableName) {
		return this.selectMoreData(appId, authToken, tenantId, tableName, null);
	}
	
	@Override
	public String selectMoreData(String appId, String authToken, String tenantId, 
			String tableName, String filterJson) {
		if(ValidateUtil.isNull(appId, authToken, tenantId)){
            throw new WebApplicationException(Response.status(Status.BAD_REQUEST).build());
		}
		
		List<HbasePut> puts = new ArrayList<HbasePut>();
		HTableInterface table = this.getPool().getTable(tableName);
		
		Scan s = new Scan();
		List<Filter> filters =  new ArrayList<Filter>();
		
		if(filterJson!=null && !"".equals(filterJson.trim())){
			List<FilterBean> filterBeans = null;
			try{
				filterBeans = JsonParserUtil.getObjectsFromJson(filterJson, FilterBean.class);
			}catch(Exception e){
				log.error(e.getMessage());
				throw new WebApplicationException(Response.status(Status.BAD_REQUEST).build());
			}
			
			for(FilterBean filterBean : filterBeans) {
				if(FilterBean.FILTER_SIGN_ROW.equalsIgnoreCase(filterBean.getSign())){
					Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,  
							new RegexStringComparator(".*"+filterBean.getKey()+":"+filterBean.getValue()+".*"));
					filters.add(filter);
				}else if(FilterBean.FILTER_SIGN_COLUMN.equalsIgnoreCase(filterBean.getSign())){
					Filter filter = new SingleColumnValueFilter(Bytes.toBytes(DEFAULT_FAMILYNAME), 
							Bytes.toBytes(filterBean.getKey()), CompareOp.EQUAL, Bytes.toBytes(filterBean.getValue()));
					filters.add(filter);
				}else{
					log.error("Didn't find filter sign "+filterBean.getSign()+".");
					throw new WebApplicationException(Response.status(Status.BAD_REQUEST).build());
				}
			}
		}
		
		//添加租户过滤
		Filter tenatFilter = new SingleColumnValueFilter(Bytes.toBytes(DEFAULT_FAMILYNAME), 
				Bytes.toBytes(this.TENANTID), CompareOp.EQUAL, Bytes.toBytes(tenantId));
		filters.add(tenatFilter);
		
		//加入过滤查询
		FilterList filterList = new FilterList(filters);
		s.setFilter(filterList);
		
		// HTable提供了个getScanner方法来完成批量查询
		ResultScanner ss = null;
		try{
			ss = table.getScanner(s);
			for (Result r : ss) {
				HbasePut put = convertor(r);
				put.setTableName(tableName);
				puts.add(put);
			}
		}catch(Exception e){
			log.error(e.getMessage());
			throw new WebApplicationException(Response.status(Status.INTERNAL_SERVER_ERROR).build());
		}finally{
			if(ss!=null){
				ss.close();
			}
			
			try {
				putTable(table);
			} catch (IOException e) {
				log.error(e.getMessage());
			}
		}

		return JsonParserUtil.getJsonFromObject(puts);
	}
	
	/**
	 * 对象转换
	 * 
	 * @param r
	 * @return
	 */
	private HbasePut convertor(Result r) {
		HbasePut put = new HbasePut();
		Map<String, String> puts = put.getKeyValues();
		if(puts == null) {
			puts = new HashMap<String, String>();
		}
		
		String row;
		//String family;
		String qualifier;
		String value;
		for (KeyValue kv : r.raw()) {
			row = new String(kv.getRow());
			//family = new String(kv.getFamily());
			//log.info("Family name is " + family);
			
			if(null == put.getRowKey()) {
				put.setRowKey(row);
			}else {
				if(!put.getRowKey().equals(row)){
					throw new RuntimeException("同一组数据中Row或Family不相等");
				}
			}
			
			qualifier = new String(kv.getQualifier());
			value = new String(kv.getValue());
			
			puts.put(qualifier, value);
		}
		put.setKeyValues(puts);
		
		return put;
	}
	
	private List<HbasePut> selectRowKeysFamilyColumn(String tableName,
			String family, String column, String...rowKeys) throws IOException {
		HTableInterface table = this.getPool().getTable(tableName);
		
		List<Get> gets = new ArrayList<Get>();
		for(String rowKey: rowKeys){
			Get g = new Get(rowKey.getBytes());
			if(family!=null) {
				if(column!=null) {
					g.addColumn(family.getBytes(), column.getBytes());
				}else{
					g.addFamily(Bytes.toBytes(family));
				}
			}
			gets.add(g);
		}
		
		List<HbasePut> puts = new ArrayList<HbasePut>();
		Result[] rs = table.get(gets);
		if(rs[0].raw().length>0){
			for(int i=0;i<rs.length;i++){
				HbasePut put = convertor(rs[i]);
				put.setTableName(tableName);
				puts.add(put);
			}
		}
		
		putTable(table);
		return puts;
	}
	
	@SuppressWarnings("deprecation")
	private void putTable(HTableInterface table) throws IOException {
		this.getPool().putTable(table);
	}
	
}
